#include "surveyor_respondent.hpp"
#include "nng/nng.h"
#include "nng/protocol/survey0/survey.h"
#include "nng/protocol/survey0/respond.h"
#include "utils/log.hpp"
#include <fmt/format.h>
#include <app/app_base.hpp>
#include <dexode/EventBus.hpp>
#include <event/events.hpp>

#define _STRINGIFY(X) #X
#define STRINGIFY(X) _STRINGIFY(X)
#define SERVER_PORT 9721
#define SERVER_BIND "tcp://*:" STRINGIFY(SERVER_PORT)

namespace SurveyorRespondent
{

    nng_socket surveyor_sock NNG_SOCKET_INITIALIZER;
    nng_aio *surveyor_aio = nullptr;

    std::string last_addr;
    nng_socket respondent_sock NNG_SOCKET_INITIALIZER;
    nng_dialer respondent_dialer NNG_DIALER_INITIALIZER;
    nng_aio *respondent_aio = nullptr;

    volatile bool stopped = false;
    volatile bool surveyoring = false;

    inline void surveyor_receive_callback(void *)
    {
        int rv = nng_aio_result(surveyor_aio);
        if (rv == 0)
        {
            auto msg = nng_aio_get_msg(surveyor_aio);
            std::string content;
            content.resize(nng_msg_len(msg),'\0');
            memcpy(&content[0], nng_msg_body(msg), content.length());
            App->Bus()->postpone(AppEvent::RespondEvent{std::move(content)});
            nng_msg_free(msg);
            // next round
            if (!stopped)
                nng_recv_aio(surveyor_sock, surveyor_aio);
        }
        else if (rv == NNG_ETIMEDOUT)
        {
            surveyoring = false;
        }
        else
        {
            L_ERROR("surveyor complete with code %d, msg %s", rv, nng_strerror(rv));
        }
        
    }

    inline void respondent_receive_callback(void *)
    {
        int rv = nng_aio_result(respondent_aio);
        if (rv == 0)
        {
            auto msg = nng_aio_get_msg(respondent_aio);
            char *data = (char*)nng_msg_body(msg);
            char *end = data + nng_msg_len(msg);
            char *split = strchr(data, ':');
            if (split != NULL)
            {
                App->Bus()->postpone(AppEvent::SurveyEvent{
                    std::string{data, split},
                    std::string{split+1, end},
                });
            }
            nng_msg_free(msg);
        }
        else
        {
            if (rv != NNG_ETIMEDOUT)
                L_ERROR("repondent complete with code %d, msg %s", rv, nng_strerror(rv));
        }
        // next round
        if (!stopped && rv != NNG_ECANCELED)
            nng_recv_aio(respondent_sock, respondent_aio);
    }

    void config_surveyor (bool enabled)
    {
        if (enabled)
        {
            if (surveyor_sock.id == 0)
            {
                int ret;
                ret = nng_surveyor0_open(&surveyor_sock);
                if (ret != 0)
                {
                    L_ERROR("Surveyor open failed: %s", nng_strerror(ret));
                }
                else
                {
                    nng_socket_set_ms(surveyor_sock, NNG_OPT_RECVTIMEO, 1000); // 1s
                    ret = nng_listen(surveyor_sock, SERVER_BIND, nullptr, 0);
                    if (ret != 0)
                    {
                        L_ERROR("Surveyor listen failed: %s", nng_strerror(ret));
                        nng_close(surveyor_sock);
                        surveyor_sock = NNG_SOCKET_INITIALIZER;
                    }
                    else
                    {
                        nng_socket_set_ms(surveyor_sock, NNG_OPT_SURVEYOR_SURVEYTIME, 2500);
                        nng_aio_alloc(&surveyor_aio, surveyor_receive_callback, nullptr);
                        nng_aio_set_timeout(surveyor_aio, 5000);
                    }
                }
            }
        }
        else
        {
            if (surveyor_sock.id != 0)
            {
                nng_aio_cancel(surveyor_aio);
                nng_aio_wait(surveyor_aio);
                nng_aio_free(surveyor_aio);
                surveyor_aio = nullptr;
                nng_close(surveyor_sock);
                surveyor_sock = NNG_SOCKET_INITIALIZER;
            }
        }
    }

    void clean_up ()
    {
        stopped = true;
        config_surveyor(false);
        config_respondent({});
    }

    void config_respondent (const std::string &ip)
    {
        if (last_addr != ip)
        {
            // close first
            if (respondent_sock.id)
            {
                nng_aio_cancel(respondent_aio);
                nng_aio_wait(respondent_aio);
                nng_aio_free(respondent_aio);
                respondent_aio = nullptr;
                nng_close(respondent_sock);
                respondent_sock = NNG_SOCKET_INITIALIZER;
                last_addr.clear();
            }

            // recreate
            if (!ip.empty())
            {
                int ret;
                ret = nng_respondent0_open(&respondent_sock);
                if (ret != 0)
                {
                    L_ERROR("Respondent open failed: %s", nng_strerror(ret));
                }
                else
                {
                    nng_socket_set_ms(respondent_sock, NNG_OPT_RECVTIMEO, 1000);
                    nng_socket_set_ms(respondent_sock, NNG_OPT_RECONNMINT, 1500);
                    nng_socket_set_ms(respondent_sock, NNG_OPT_RECONNMAXT, 0);
                    std::string server_addr = fmt::format("tcp://{}:{}", ip, SERVER_PORT);
                    ret = nng_dialer_create(&respondent_dialer, respondent_sock, server_addr.c_str());
                    if (ret != 0)
                    {
                        L_ERROR("Respondent dialer create failed: %s", nng_strerror(ret));
                        nng_close(surveyor_sock);
                        surveyor_sock = NNG_SOCKET_INITIALIZER;
                    }
                    else
                    {
                        // 3秒重试
                        ret = nng_dialer_set_ms(respondent_dialer, NNG_OPT_RECONNMINT, 1500);
                        ret = nng_dialer_set_ms(respondent_dialer, NNG_OPT_RECONNMAXT, 0);
                        ret = nng_dialer_start(respondent_dialer, NNG_FLAG_NONBLOCK); // 好像会等到连接成功才会返回？

                        ret = nng_aio_alloc(&respondent_aio, respondent_receive_callback, nullptr);
                        nng_recv_aio(respondent_sock, respondent_aio);
                        last_addr = ip;
                    }
                }
            }
        }
    }
    
    void surveyor_range(const std::string &date_start, const std::string &date_end)
    {
        if (surveyor_sock.id)
        {
            surveyoring = true;
            auto data = fmt::format("{}:{}", date_start, date_end);
            int ret = nng_send(surveyor_sock, &data[0], data.length(), 0);
            if (ret != 0)
            {
                surveyoring = false;
                L_ERROR("surveyor_range error: %s", nng_strerror(ret));
            }
            else
            {
                // 启动异步接收
                nng_recv_aio(surveyor_sock, surveyor_aio);
            }
        }
    }
    
    void respondent_data(const std::string &content)
    {
        nng_msg *msg;
        nng_msg_alloc(&msg, content.size());
        memcpy(nng_msg_body(msg), content.data(), content.length());
        nng_sendmsg(respondent_sock, msg, 0);
    }
    
    bool is_surveying()
    {
        return surveyoring;
    }

}  // namespace SurveyorRespondent
